package cn.dfun.sample.flink.tabletest

import cn.dfun.sample.flink.apitest.SensorReading
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.types.Row

/**
  * 1.数据转换操作
  * 2.sink到文件
  */
object FileOutputTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv = StreamTableEnvironment.create(env)

    val inputPath = "C:\\wor\\flink-sample\\src\\main\\resources\\sensor"
    tableEnv.connect(new FileSystem().path(inputPath))
      // 旧版,非标,弃用不支持kafka
      //        .withFormat(new OldCsv())
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temp", DataTypes.DOUBLE())
      )
      .createTemporaryTable("inputTable")
    // 3 转换操作
    val sensorTable = tableEnv.from("inputTable")
    // 3.1 简单转换
    val resultTable = sensorTable
        .select('id, 'temp)
        .filter('id === "sensor_1")
    // 3.2 聚合转换
    val aggTable = sensorTable
        .groupBy('id) // 基于id分组
        .select('id, 'id.count as 'count)
//    resultTable.toAppendStream[(String, Double)].print("result")
//    aggTable.toRetractStream[(String, Long)].print("agg")
    // upsert模式在流中无法直接做转换
//    aggTable.toRetractStream[Row].print("agg")

    // 4 sink到文件
    // 注册输出表
    val outputPath = "C:\\git\\samples\\flink-sample\\src\\main\\resources\\sensor.txt"
    tableEnv.connect(new FileSystem().path(outputPath))
      // 旧版,非标,弃用不支持kafka
      //        .withFormat(new OldCsv())
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("temp", DataTypes.DOUBLE())
//          .field("cnt", DataTypes.BIGINT())
      )
      .createTemporaryTable("outputTable")
    // todo 输出成文件夹
    // 文件系统做了更新操作不能写入
    resultTable.insertInto("outputTable")

    env.execute("file output test")
  }
}
